得物自动化平台执行器设计与实现

目录

一、名词解释

    1. 核心词解释

    2. 只读数据保护机制

二、平台化Case业务架构

三、计划执行器

四、用例&组件调试器

五、工作器（Worker）

    1. 脚本组件工作器

    2. 用例工作器

六、系统方法

    1. 系统方法库

    2. 系统方法实现

    3. 系统方法代理

七、占位符

    1. 普通占位符

    2. 增强占位符

    3. 占位符使用案例

    4. 核心代码片段

八、内置库

九、总结

随着公司接口自动化应用逐渐深入，老自动化方案弊端日渐凸显（线下脚本&自动化框架 + Jenkins + 平台[调度 + 报表 + ...]），如：技术栈&框架&三方库差异大、用户兼容性差、用例编写效率低、平台接入复杂、平台化适配性差、用例脚本不可控、用例维护成本高、执行耗时长等。为此我们将自动化平台由“半平台化”转型为“全平台化”，实现了轻量高效、功能完备、使用简单、标准化程度高的自动化平台，支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例。在用例执行方面，新平台没有被传统的自动化框架所束缚，自研了更适合平台化的“自动化用例执行器”。

自动化执行器是自动化平台自研的自动化用例执行器，负责具体执行平台编写的自动化用例和脚本，支持单独调试和按测试计划批量执行用例。主要提供串/并行跑用例、占位符、系统方法、环境变量(只读)、变量空间(读/写)、解释执行API.步骤、原生执行代码脚本等能力。

执行器是参考了优秀接口测试工具（Jmeter、Postman、eolink、MeterSphere等）和主流单元测试框架（TestNG、PyTest、unittest等）后进行自主研发的“自动化用例-执行器”（Made in 得物）。

一

名词解释

核心词解释

• 环境变量：用户常用配置(域名、库链接、Redis 链接、自定义配置、数据驱动配置等)，系统注入变量（运行环境、染色标、触发人、任务 ID 等）【用户态(只读)】。

• 变量空间：系统、用户存取变量的容器，主要用于用例的通信，分为“全局空间(g_vars)”和“局部空间(l_vars)”【用户态读/写】，变量空间会被绑定到对应的工作线程上方便实时读取。

• 系统变量：由系统主动注入到变量空间中的变量，以“_”开头【用户态(只读)】。

• 系统方法：由系统提供的工具类方法如：生成随机数、处理时间等常用方法。

• 断言方法：由系统提供的断言方法，断言时只能用此类方法，否则无法监听断言结果。

• 占位符：用于在页面上进行参数替换，支持从变量空间、环境变量中取值，或调用系统方法，支持语法表达式。

• 控制区域：用于区分用户可操作的区域，类似于操作系统的用户态和内核态。

• 执行器控制区域（内核态）：除了用户态，都是内核态。

• 用户控制区域（用户态）：用户可以写脚本的地方，用户可以填参数的地方。

• 用户态/内核态示意图。

• 脚本属于用户态。

• 请求体、接口 PATH 等也属于用户态。

只读数据保护机制

由于环境变量、系统变量、组件调用参数等信息都是基础的元数据，不能被用户态随意更改，否则会引发不可预知的问题，所以用户态操作元数据时必须加一层保护机制，保护机制如下：

用户态通过代理方式交互：

• 用户态通过代理操作执行器提供的数据。

• 环境变量代理（EnvVarsProxy）：只提供 get()方法，不提供 set() 和  remove()方法。

• 动态绑定拦截：解决动态语言特性问题。

class EnvVarsProxy(object):
    """
    环境变量代理
        - 可读不可写
    """
    def __init__(self, env_vars_space: EnvVarsSpace):
        self.__env_vars_space = env_vars_space

    def __setattr__(self, name, value):
        # 拦截非法调用栈（保护关键数据）
        stacks = inspect.stack()
        source_stack: FrameInfo = stacks[1]
        if source_stack.filename != __file__:
            raise Exception("EnvVarsProxy 对象不可设置属性")

        super().__setattr__(name, value)

    def __getattribute__(self, name):
        if name == '_EnvVarsProxy__env_vars_space' or name == '__dict__':
            # 拦截非法调用栈（保护关键数据）
            stacks = inspect.stack()
            source_stack: FrameInfo = stacks[1]
            if source_stack.filename != __file__:
                raise Exception("EnvVarsProxy 对象不可访问受保护属性")

        return super().__getattribute__(name)

• 深层保护-引用类型数据保护：get 数据时使用深拷贝方式返回数据副本，即使用户态改了数据副本的内部信息，元数据中的信息不会被影响，则不会影响其他用例或组件的正常执行。

def get(self, key):
    value = self.__env_vars_dict.get(str(key))
    if value is not None:
        return copy.deepcopy(value)

    return value

• 变量空间代理（VarsProxy）：非系统变量不限制，拦截系统变量 set 和 remove 操作。

• sys_funcs.get_call_param()：只提供本方法获取调用参数，返回深拷贝的副本数据。

二

平台化Case业务架构

自动化-平台侧

自动化-执行器

三

计划执行器

计划执行器是用于正式执行测试用例的执行器，负责前置加载任务需要的所有资源，包括用例、组件、环境变量等资源。然后按测试计划的用例编排方式，执行具体的用例，具体内容如下：

• 系统变量注入。

• 环境变量注入。

• 脚本前置写入：必须在主线程中前置写入，否则出现偶现无解的灵异现象。

• 脚本工作空间隔离设置：用于控制脚本中操作文件时的默认工作目录。

• 公共组件匹配：负责对公共组件进行引用匹配（用例中添加组件实际上是个拓展指针，指向组件实例）。

• 数据驱动：驱动变量注入【针对非常规项目】。

• Dubbo 服务：开启 Dubbo 服务发现任务。

• 任务运行简报：开启任务运行简报上报任务，让用户可以看到运行时的用例运行状态。

• 执行计划前置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。

• 按批次并行执行用例组：批次间并行执行，批次内运行模式控制串/并行执行用例。

• 批次内拆分子批次-强制串行用例拆分到新的子批次。

• 常规用例-单条执行：触发用例工作器（CaseWorker）执行用例。

• 数据驱动用例-裂变为多条顺序执行：触发用例工作器（CaseWorker）执行用例。

• 执行计划后置脚本：触发脚本组件工作器（ScriptComponentWorker）执行“脚本组件”。

• 执行结果上报：由“报告收集器”根据“日志管理器”汇总每条用例的执行结果以及错误日志和过程日志，并收集接口调用记录，一同上报到自动化平台，根据情况自动触发分批或全量上报结方式。

四

用例&组件调试器

主要用于对用例或者组件进行实时调试，调试结果不落库，不会影响用例的成功率。调试器是精简版的计划执行器，只针对单个用例或者组件进行单条调试，相对更加轻量级。

五

工作器（Worker）

负责控制执行具体的用例和脚本组件，以及局部空间(l_vars)间初始化与回收、系统变量注入等，主要包括“ScriptComponentWorker”和“CaseWorder”等工作器。

脚本组件工作器

脚本组件工作器（ScriptComponentWorker）负责控脚本组件的执行，动态绑定主步骤环境变量和归属项目环境变量，构造脚本组件(ScriptComponent)、柔性初始化局部空间(l_vars)、关键系统变量注入等。

• 由于脚本组件只会在主线程和用例线程中被执行，所以归属的执行线程空间中若已经存在局部空间，则直接复用。

def _init(self):
    """初初始"""
    self.component = ScriptComponent(self.task_md, self.raw_data, self)
    # 存储用例关键信息到局部变量的系统变量中
    # 若当前线程已绑定VarsGroup对象，则直接返回其中VarsSpace对象，否则进行初始化然后返回 
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    l_vars_space.set(VarsSpaceManager.SCRIPT_COMPO_ID_KEY, self.component.component_md.component_id)
    l_vars_space.set(VarsSpaceManager.SCRIPT_COMPO_NAME_KEY, self.component.component_md.name)
def _clear(self):
    """空间清理"""
    ScriptComponentUtilImp.set_call_param(None)  # 清空当前组件的参数调用
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    l_vars_space.remove(VarsSpaceManager.SCRIPT_COMPO_ID_KEY)
    l_vars_space.remove(VarsSpaceManager.SCRIPT_COMPO_NAME_KEY)

脚本组件（ScriptComponent）执行逻辑片段

通过反射动态导入执行脚本模块，然后进行组件调用检验&参数替换（替换入参中的占位符），动态绑定组件调用参数（用户态只读），然后调用脚本的 call 方法【def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):】。

def run_script(self):
    ...
    script_obj = importlib.import_module(script_model_path)  # 动态导入模块
    ...
    # 组件相关代理对象准备，组件内只能通过代理对象与执行器进行脚本，代理对象会进行安全拦截 
    env_vars_proxy: EnvVarsProxy = EnvVarsSpaceManager.get_env_vars_proxy(self.component_md.project_id)
    g_vars_proxy: VarsProxy = VarsSpaceManager.get_g_vars_proxy()
    l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()
    g_vars_space: VarsSpace = VarsSpaceManager.get_g_vars_space()
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    sys_funcs_proxy: SysFuncsProxy = SysFuncsManager.get_sys_funcs_proxy()
    asserts_proxy: AssertFuncsProxy = AssertManage.get_sys_funcs_proxy()

    # call_param参数的占位符替换 
    new_call_param = {}
    for _key, _value in self.component_md.call_param.items():
        if isinstance(_value, str):
            sub_code, _value = VarExtraction.param_substitute(
                self.component_worker.logger_group, g_vars_space, l_vars_space, env_vars_proxy, _value,
                fail_desc_prefix=f'{self.component_desc_info}替换调用参数')
            if not sub_code:
                return

        # 兼容老的数据
        elif isinstance(_value, (dict, list)):
            _ = json.dumps(_value, ensure_ascii=False)
            sub_code, _ = VarExtraction.param_substitute(
                self.component_worker.logger_group, g_vars_space, l_vars_space, env_vars_proxy, _,
                fail_desc_prefix=f'{self.component_desc_info}替换调用参数')
            if not sub_code:
                return

            # 还原类型-转换失败则放弃转换（减少组件报错率）
            try:
                _value = json.loads(_)
            except Exception as e:
                self.component_worker.logger_group.logger_proxy.warning(f'替换调用参数失败-替换成功后不符合JSON格式【{str(e)}】',
                                                                        _add_to_logger_space=False)
        new_call_param[_key] = _value

    # 对调用参数类型进行兜底转换-转换失败则放弃转换（减少组件报错率）
    for rule in self.component_md.param_rule:
        _key = rule['name']
        if _key in new_call_param and rule['type'] != 'string':
            _value = new_call_param[_key]
            if isinstance(_value, str):
                try:
                    new_call_param[_key] = json.loads(_value)
                except Exception as e:
                    self.component_worker.logger_group.logger_proxy.warning(f'替换调用参数失败-类型不匹配【{str(e)}】',
                                                                            _add_to_logger_space=False)

    # 动态绑定组件调用参数，组件中可通过 sys_funcs.get_call_param() 获取到调用参数 
    ScriptComponentUtilImp.set_call_param(new_call_param)
    try:
        script_obj.call(env_vars_proxy, g_vars_proxy, l_vars_proxy, sys_funcs_proxy, asserts_proxy,
                        self.component_worker.logger_group.logger_proxy)
    except AssertFail as e:
    ...

用例工作器

用例工作器（CaseWorder）负责控制用例执行，强制初始化局部空间(l_vars)、主项目环境变量绑定、构造测试用例(DataCase)、失败重试控制（用例维度）。

def run(self):
    for i in range(self._retry_count):
        self._set_case_status(False)
        if i > 0:
            # 等待 retry_sleep 秒后进行重试
            if self.data_case.data_case_md.retry_sleep > 0:
                time.sleep(self.data_case.data_case_md.retry_sleep)

            self.retry_i = i
            self.logger_group.logger_proxy.info(
                f'第 {i} 次重试用例，初始化日志组（已等待 {self.data_case.data_case_md.retry_sleep} 秒）',
                _add_to_logger_space=False)
            # 重置当前用例日志存储对象（实现失败重试日志隔离）
            self.logger_group = LoggerManager.init_logger_group(
                self.task_md, f'{self.task_md.runtime_type},case_id:{self.raw_data["son_step_id"]}')

        # 标记是否继续循环
        if_continue = False
        try:
            # 执行用例，失败则根据重试次数进行重试，成功则结束循环
            self._run()
            # 若存在执行错误日志则表示执行失败，重试执行
            if self.logger_group.logger_space.fail_msg_records:
                self._set_case_status(True)
                if_continue = True

        except AssertFail as e:
            # 这里获取到断言异常只可能是用例前置脚本的断言异常
            self.logger_group.logger_space.add_fail_msg(
                LogUnit(title=self.data_case.case_desc_info,
                        content=f'断言失败: {str(e)}',
                        newline=1))
            self.logger_group.logger_proxy.info(f'执行用例步骤-{str(e)}', _add_to_logger_space=False)
            self._set_case_status(True)
            self._run_after_script()
            continue
        ...
def _run(self):
    self._init()
    self.data_case.run_case()
def _init(self):
    self._fail_msg = ''
    self.data_case = DataCase(self.task_md, self.raw_data, self)
    # 每次运行都需要强制初始化，确保失败重试不被影响 
    VarsSpaceManager.init_l_vars_proxy()
    # 设置主步骤所在项目的项目ID
    EnvVarsSpaceManager.set_main_step_project_id(self.data_case.data_case_md.project_id)

测试用例（DataCase）执行逻辑片段

• 执行测试用例。

def run_case(self):
    """
    执行测试用例
        - 执行脚本组件，复用 ScriptComponentWorker 的能力     
        - case为失败状态时会执行剩下的有"is_finally"标的步骤    
            - "is_finally"标的步骤只有是断言失败时才影响case结果
    """
    l_vars_space: VarsSpace = VarsSpaceManager.get_l_vars_space()
    # 存储用例关键信息到局部变量的系统变量中
    l_vars_space.set(VarsSpaceManager.CASE_NAME_KEY, self.data_case_md.name)
    l_vars_space.set(VarsSpaceManager.CASE_ID_KEY, self.data_case_md.id)
    l_vars_space.set(VarsSpaceManager.CASE_AUTHOR_KEY, self.data_case_md.create_user)
    ...
    # 设置驱动数据到局部空间
    if self.case_worker.is_derive_case:
        drive_i_values: dict = self.case_worker.drive_info.get_drive_i_values(self.case_worker.drive_i)
        l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()
        for k, v in drive_i_values.items():
            # 替换驱动数据的值（替换占位符中的变量）
            sub_code, _value = self.param_substitute(v, f'{self._case_desc_info}替换驱动参数"{k}"的值')
            if not sub_code:
                return False
            l_vars_proxy.set(k, _value)

    res_code = self.run_before_script()
    if not res_code:
        return

    for case_step in self.data_case_md.case_steps:
        if case_step.step_type == DataCaseMd.API_H_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_http_step(case_step)
        elif case_step.step_type == DataCaseMd.API_D_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_dubbo_step(case_step)
        elif case_step.step_type == DataCaseMd.SCRIPT_COMPONENT_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            # 执行脚本组件
            self._run_script_component(case_step)
        elif case_step.step_type == DataCaseMd.API_G_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_grpc_step(case_step)
        elif case_step.step_type == DataCaseMd.SCRIPT_STEP_TYPE:
            if self.is_fail and not case_step.is_finally:
                continue
            self._run_script_step(case_step)

执行 API.*步骤：

• 以 API.H(HttpStep)步骤为例，API.D(DubboStep)、API.G(GRpcStep) 步骤的主要逻辑基本相同。

def _run_http_step(self, case_step: HttpStepMd):
    from .steps.http_step import HttpStep
    retry_count = int(case_step.retry_count or 0) + 1 if self.task_md.allow_retry else 1
    retry_sleep = int(case_step.retry_sleep or 0)

    # 执行结果标
    run_res = False
    http_step = None
    for i in range(retry_count):
        if i > 0:
            if retry_sleep > 0:
                time.sleep(retry_sleep)

            # 转移错误日志到执行记录中 
            self.case_worker.logger_group.logger_space.move_fail_msg()

        http_step = HttpStep(self, case_step, retry_i=i)
        run_res = http_step.run_step()
        if run_res:
            return

    if not run_res:
        if not case_step.is_finally:
            self.is_fail = True
        elif http_step.assert_fail:
            # finally 步骤如果是断言失败，才会影响用例结果 
            self.is_fail = True
        else:
            # finally 步骤如果不是断言失败，则不影响用例结果，需要转移错误日志到执行记录中 
            self.case_worker.logger_group.logger_space.move_fail_msg()

• http 步骤执行内容：

• 准入条件校验。

• 执行步骤的“前置脚本”。

• 域名替换。

• 参数替换：替换“api host、api path、params、headers、body、proxies、asserts”数据中的占位符变量。

# 替换headers参数
new_headers = {}
for _header in self.http_step_md.headers:
    if _header.get('is_use') == '1' and _header.get('key_'):
        _value = _header.get('value') or ''
        sub_code, _value = self.data_case.param_substitute(str(_value), f'{self._step_desc_info}替换headers参数: ')
        if not sub_code:
            return False
        new_headers[str(_header['key_'])] = _value

• 接口调用：负责发送 http 请求。

def send_request(self, host: str, url: str, full_url: str, data: str = '', params=None, headers=None, timeout=None,
                 proxies=None)
 -> Tuple[bool, Any]:
    # 请求头追加字段
    headers = self._handler_headers(headers)
    ...
    # 记录请求体数据
    storage_body = data
    try:
        ...
        if http_method in ['POST', 'PATCH', 'PUT', 'DELETE']:
            # 处理请求体数据格式
            handler_code, new_data, storage_body = self._handler_request_body(data, headers)
            if not handler_code:
                return False, None

            # 添加sign信息并更新url
            if self._sign_info is not None:
                self._api_request_response['full_url'] = full_url = self._url_add_sing(full_url, params)
                self._logger_space.add_log(LogUnit(title='追加sign', content=full_url, indent=1),
                                           log_type=LogInfo.DEBUG_LOG)

            self._api_request_response['request_body'] = storage_body

            if http_method == 'POST':
                resp = self._send_post(host, url, data=new_data, params=params, headers=headers, timeout=timeout,
                                       proxies=proxies)
            elif http_method == 'PATCH':
                resp = self._send_patch(host, url, data=new_data, params=params, headers=headers, timeout=timeout,
                                        proxies=proxies)
            elif http_method == 'PUT':
                resp = self._send_put(host, url, data=new_data, params=params, headers=headers, timeout=timeout,
                                      proxies=proxies)
            else:
                resp = self._send_delete(host, url, data=new_data, params=params, headers=headers, timeout=timeout,
                                         proxies=proxies)

        elif http_method == 'GET':
            resp = self._send_get(host, url, params=params, headers=headers, timeout=timeout, proxies=proxies)
     ...

• 断言结果：按步骤中配置的断言项进行批量断言。

• 期望值为数值类型时值 支持数学运算 如下案例:

• 图中期望值为 int 类型，期望值表达式为【${_l->goods1->price} + 5 - ${_e->xxx->goods2->price}】。

• 其中${_l->goods1->price} ：代表取出局部空间中的“goods1”下面的“price”的值，若为10。

• 其中${_e->xxxx->goods2->price} ：代表取出环境变量中的“xxx”下面的“goods1”下面的“price”的值，若为3。

• 则占位符变量替换后得到结果为【10 + 5 - 3】。

• 数学运算后得到最终结果为【12】。

• 最终将断言接口返回数据中的 $.data.price 是否 等于 12。

expect_value_type = str(assert_item.get('expect_value_type'))
expect_value_func = self.EXPECT_VALUE_TYPE_DICT.get(expect_value_type)
...
# 期望值类型转换
try:
    expect_value = expect_value_func(assert_item.get('expect_value'))
except Exception as e:
...

def int_calculation(value: str):
    """int类型转换，支持数学运行"""
    if isinstance(value, int):
        return value

    value = str(value)
    if value.isnumeric():
        return int(value)

    value = ast.literal_eval(value)
    return int(value)


def float_calculation(value: str):
    """float类型转换，支持数学运行"""
    if isinstance(value, float):
        return value

    value = str(value)
    if value.isnumeric():
        return float(value)

    value = ast.literal_eval(value)
    return float(value)

• 提取参数：将接口的返回数据按提取规则进行提取并存储到对应的变量空间中。

• 执行步骤的“后置脚本”。

六

系统方法

由执行器提供的工具方法，供用户态通过代理（SysFuncsProxy）进行使用的工具类方法，可以在占位符和脚本代码中使用，如：生成随机数、处理时间等方法。

系统方法库

装载系统方法到方法库，装载后才可以在用户态的脚本和占位符中使用该方法。

@classmethod
def load_sys_func(cls, name='', param_rules: List[ParamRule] = None):
    """
    装载系统方法
    :param name: 方法名
    :param param_rules: 参数规则（注意：缺省值参数放到最后）
    """
    def decorator(func):
        _name = name or func.__name__
        if name in cls.SYS_FUNCS_DICT:
            raise SysFuncsRepeatError(f'系统方法"{name}"重复')

        cls.SYS_FUNCS_DICT[_name] = {
            "func": func,
            "param_rules": param_rules or []
        }
        return func

    return decorator

系统方法实现

实现系统方法，需要通过@SysFuncsLib.load_sys_func()注入方法相关信息，以便调用规范校验和快捷模版生成。

• 案例：转换为时间戳(switch_timestamp)，该方法接收 3 个参数，且都是非缺省值参数。

@SysFuncsLib.load_sys_func(name='switch_timestamp', param_rules=[ParamRule('length', int, required=True),
                                                                 ParamRule('date_string', str, required=True),
                                                                 ParamRule('format', str, required=True)])
def switch_timestamp(length: int, date_string: str, format: str) -> int:
    """
    转换为时间戳
        - 将日期字符串转换为时间戳
    :param length: 位数
    :param date_string: 日期字符串
    :param format: 日期格式
    """
    date_obj = datetime.datetime.strptime(date_string, format)
    timestamp = str(date_obj.timestamp()).replace('.', '')

    # 补位
    n = len(timestamp)
    if n < length:
        timestamp += '0' * (length - n)

    return int(timestamp[: length])

系统方法代理

class SysFuncsProxy(object):
    """系统方法代理"""
    ...
    def __getattr__(self, name):
        func_info = SysFuncsLib.SYS_FUNCS_DICT.get(name)
        if not func_info:
            raise SysFuncsNonExistentError(f'系统方法"{name}"不存在')

        return func_info['func']

    ...

系统方法在脚本中使用简介

• 执行器控制区域（内核态）：系统负责获取到（用户态）需要的所有代理对象，并触发调用脚本入口方法。

...
# 相关代理对象准备 
env_vars_proxy: EnvVarsProxy = EnvVarsSpaceManager.get_env_vars_proxy(self.data_case.data_case_md.project_id)
g_vars_proxy: VarsProxy = VarsSpaceManager.get_g_vars_proxy()
l_vars_proxy: VarsProxy = VarsSpaceManager.get_l_vars_proxy()
sys_funcs_proxy: SysFuncsProxy = SysFuncsManager.get_sys_funcs_proxy()
asserts_proxy: AssertFuncsProxy = AssertManage.get_sys_funcs_proxy()
# 调用入口方法 
script_obj.call(env_vars_proxy, g_vars_proxy, l_vars_proxy, sys_funcs_proxy,
                asserts_proxy, self._logger_proxy)

...

• 用户控制区域（用户态）：在脚本代码中使用系统方法。

import json
import requests

def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    # 使用系统方法 
    res = sys_funcs.switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')
    logger.info(res)

系统方法在占位符中使用介绍

• 用户控制区域（用户态）：在页面数据区域使用系统方法（如在请求体参数中使用）。

{
    "timestamp": ${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')},
    "xxx": "yyy"
}

• 执行器控制区域（内核态）：系统负责提取原始数据中的方法调用占位符，解析调用语义，执行调用并替换结果。

• 主要包括：

• 方法识别。

• 调用方式校验：无参调用、有参调用。

• 入参规则校验。

• 从字符串中解析出对应的入参。

• 参数个数、参数类型校验、必填参数校验。

• 调用系统方法 & 结果替换。

• 系统方法占位符替换代码片段：

...
# 获取系统方法
sys_func = sys_funcs_proxy.get_func(func_name)
...
# 获取系统参数规则
param_rules: List[ParamRule] = sys_funcs_proxy.get_param_rules(func_name)
# 无参调用 
if not param_rules:
    try:
        value = sys_func()
    except BaseException as e:
        ...
# 有参调用 
else:
    args_str: str = key[len(func_name) + 2:-1]  # 提取入参
    if not args_str:
        ...
        return False, ''
    raw_args = [str(_).strip() for _ in args_str.split(',')]
    raw_args_n = len(raw_args)

    # 判断参数总个数正确性
    if raw_args_n > len(param_rules):
        ...
        return False, ''

    required_n = 0
    for param_rule in param_rules:
        # param_rule.required 为 bool 类型，可以参与+法运行
        required_n += param_rule.required

    # 判断必填参数个数正确
    if raw_args_n < required_n:
        ...
        return False, ''

    # 生成系统方法入参
    _args = []
    for i, param_rule in enumerate(param_rules):
        if param_rule.required and i + 1 > raw_args_n:
            ...
            return False, ''
        elif i + 1 > raw_args_n:
            # 未传部分缺省参数时结束循环
            break

        try:
            _raw_arg = raw_args[i]
            if param_rule.param_type is bool:
                # 布尔类型
                if _raw_arg.capitalize() == 'True':
                    _arg = True
                elif _raw_arg.capitalize() == 'False':
                    _arg = False
            elif param_rule.param_type is str:
                if len(_raw_arg) >= 2 and _raw_arg[0] == _raw_arg[-1] and _raw_arg[0] in ['"', "'"]:
                    _arg = _raw_arg[1:-1]
                else:
                    ...
                    return False, ''
            else:
                # 其他类型
                _arg = param_rule.param_type(_raw_arg)
        except Exception as e:
            ...
            return False, ''

        _args.append(_arg)

    try:
        value = sys_func(*_args)
    except BaseException as e:
        ...
        return False, ''

七

占位符

占位符是自动化平台使用频率很高的功能，提供页面数据区域与对应变量空间高效通信的能力。用于在页面上进行参数替换，支持从变量空间、环境变量中取值，或调用系统方法。分为普通占位符和增强占位符。

普通占位符

主要用于在页面数据区域获取变量空间中的变量值如 ${xxx}、${_case_id}，调用系统方法如 ${@get_trace_id()}。

增强占位符

支持使用语法表达式，可以准确、快捷、高效、多层级获取目标值。

• 拓展7项能力：

• 支持直接取变量的属性，“->”代表属性类似于常规代码中的“.”

• 支持直接取特殊类型(list，str，tuple)的索引和切片等复杂操作

• 支持多层级取值

• 支持取常规引用类型的属性，如：object.a

• 占位符&脚本中支持快捷取值

• 支持指定取对应空间的变量（局部、全局变量 或 环境变量）

• 拓展语法：

• _l->：指定从局部变量取值

• _g->：指定从全局变量取值

• _e->：指定从环境变量取值

• ->：取属性或key值（常规引用类型取属性，字段类型取key值）

• []：索引或切片操作（支持list，str , tuple 等类型）（与Python本身的用法类似）

• 示例：

• [2]：取索引为2的元素

• [-1]：取最后一位元素

• [1:30]: 切片-取第2个到第30个

• [1:30:2]：切片-取第2个到第30个，步长为2

• [::-1]：切片（顺序颠倒）

【页面正使用占位符】

【脚本中使用增强key】

占位符使用案例

• 案例1："demoMobile": "${@random_fix_mobile("135")}"

• 分析：存在一个普通占位符，调用系统方法random_fix_mobile("135"),该方法用于获取固定前缀随机手机号，支持用户手动传入 <= 13位的前缀，随机补充缺少的手机号位数，且整体符合手机号规范

• 替换结果："demoMobile": "13579996688"

• 案例2："demoXx": ${orders[3]->labs[2:10:2]}

• 分析：存在一个增强占位符，从局部空间或全局空间（优先局部空间）中获取订单列表(orders)中第4个订单的标签列表(labs)中第3到第10个元素步长为2

• 替换结果："demoXx": ["lab3", "lab5", "lab7", "lab9"]

• 案例3："demoYy": "你好！${@random_zh(3)}，截止到${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')}时间，你在${_env}环境购买'${_e->goods.apple->name}'商品${_l->buy_times}次"。

• 分析：存在三个普通占位符和两个增强占位符。

• ${@random_zh(3)}: 调用系统方法生成 3 个随机汉字。

• ${@switch_timestamp(13, "2022-08-22 10:10:10", '%Y-%m-%d %H:%M:%S')}: 调用系统方法将日期格式转换成 13 位时间戳。

• ${_env}: 获取系统变量中的当前执行环境。

• ${_e->goods.apple->name}: 获取环境变量中“goods.apple”变量下的“name”字段的值。

• ${_l->buy_times}：获取局部变量中“buy_times”变量的值。

• 替换结果："demoYy": "你好！觙奲琍，截止到 1661134210000 时间，你在 t1 环境购买'苹果'商品 5 次"。

• 最终结果如下：

{
  "batchNo": "1108000124800131",
  "demoMobile": "13579996688",
  "demoXx": ["lab3", "lab5", "lab7", "lab9"],
  "demoYy": "你好！觙奲琍，截止到1661134210000时间，你在t1环境购买'苹果'商品5次",
  ...
}

核心代码片段

占位符 key 表达式解析

按占位符语法将 key 解析成对应的子 key 列表，如：${orders[3][2]->labs[2:10:2]} 将被解析成 5 个子表达式，即 sub_keys = ['orders', '[3]', '[2]', 'labs', '[2:10:2]']。

@classmethod
def parsing_sub_keys(cls, key: str) -> list:
    """
    占位符的key解析成子key列表
    """
    if not key:
        return []
    
    # 缓存加速
    sub_keys = cls.KEY_MAP.get(key)
    if sub_keys is not None:
        return sub_keys

    raw_sub_keys = key.split('->')
    sub_keys = []
    # 解析表达式
    for sub_key in raw_sub_keys:
        if '[' in sub_key or ']' in sub_key:
            if len(sub_key) <= 3 or sub_key.startswith('[') or not sub_key.endswith(']'):
                raise ParseKeyException(f'表达式非法`{key}`')
            if sub_key.count('[') != sub_key.count(']'):
                raise ParseKeyException(f'表达式非法`{key}`')

            # 添加当前层的第一个字段名
            _k0 = sub_key.split('[')[0]
            sub_keys.append(_k0)

            _sub_key: str = sub_key[len(_k0):]
            if '][' not in _sub_key:
                if _sub_key.find('[') > _sub_key.find(']') or _sub_key.count('[') > 1 or _sub_key.count(']') > 1:
                    raise ParseKeyException(f'表达式非法`{key}`')
                sub_keys.append(_sub_key)
            else:
                _ = '<!-!>'  # 切割辅助符
                while _ in _sub_key:
                    _ += _  # 防止冲突

                _sub_keys = _sub_key.replace('][', f']{_}[').split(_)
                for _sub_key in _sub_keys:
                    if _sub_key.find('[') > _sub_key.find(']') or _sub_key.count('[') > 1 or _sub_key.count(']') > 1:
                        raise ParseKeyException(f'表达式非法`{key}`')
                    sub_keys.append(_sub_key)

        else:
            sub_keys.append(sub_key)

    cls.KEY_MAP[key] = sub_keys
    return sub_keys

占位符参数提取

根据提取到的子表达式（sub_keys）组层级匹配获取数据。

def get_var(self, key: str):
    """获取变量值"""
    sub_keys = self.parsing_sub_keys(key)
    if not sub_keys:
        return None

    sub_key_0 = sub_keys[0]
    if sub_key_0 == '_g':
        var, is_found = self.get_target_var(sub_keys[1:], self._g_vars, key)
    elif sub_key_0 == '_l':
        var, is_found = self.get_target_var(sub_keys[1:], self._l_vars, key)
    elif sub_key_0 == '_e':
        var, is_found = self.get_target_var(sub_keys[1:], self._env_vars, key)
    else:
        # 不指定变量空间时优先在 局部空间中查找，
        var, is_found = self.get_target_var(sub_keys, self._l_vars, key)
        if not is_found:
            var, is_found = self.get_target_var(sub_keys, self._g_vars, key)

    return var

@classmethod
def get_target_var(cls, sub_keys: list, i_var: any, raw_key: str):
    """
    获取指定空间中的变量值
    """
    is_found = False
    var = None
    i = 0
    for i, key in enumerate(sub_keys):
        is_found = False
        if i_var is None or not key:
            break

        # 获取对象里的字段or属性
        if not ('[' in key or ']' in key):
            # 字典类型
            if isinstance(i_var, dict):
                is_found = key in i_var
                i_var = i_var.get(key)

            # 获取空间中的变量值
            elif isinstance(i_var, (VarsSpace, EnvVarsProxy)):
                is_found = i_var.exist(key)
                if is_found:
                    i_var = i_var.get(key)
                else:
                    i_var = None

            # 获取其他类型对象的属性
            else:
                if hasattr(i_var, key):
                    i_var = getattr(i_var, key)
                    is_found = True
                else:
                    i_var = None

        # 获取列表、字符串、元组里的数据
        else:
            if type(i_var) not in [list, str, tuple, bytes, bytearray]:
                raise ParseKeyException(f'表达式与实际变量值类型不匹配，无法通过索引取值`{raw_key}`')

            _index_temp = key[1:-1]
            try:
                _index_temp = int(_index_temp)
            except Exception as e:
                if 0 < _index_temp.count(':') <= 2:
                    _i_temps = _index_temp.split(':')
                    for _i, _i_temp in enumerate(_i_temps):
                        try:
                            if _i_temp != '':
                                _i_temps[_i] = int(_i_temp)
                            else:
                                _i_temps[_i] = None
                        except Exception as e:
                            raise ParseKeyException(f'表达式非法`{raw_key}`，切片值只能是整数')
                    if len(_i_temps) == 2:
                        i_var = i_var[_i_temps[0]: _i_temps[1]]
                        is_found = True
                    else:
                        i_var = i_var[_i_temps[0]: _i_temps[1]: _i_temps[2]]
                        is_found = True
                else:
                    raise ParseKeyException(f'表达式非法`{raw_key}`')
            else:
                len_i_var = len(i_var)
                if 0 <= len_i_var + _index_temp < 2 * len_i_var:
                    i_var = i_var[_index_temp]
                    is_found = True
                else:
                    raise ParseKeyException(f'表达式与实际变量值不匹配取值时索引越界(list index out of range)`{raw_key}`')

    if i == len(sub_keys) - 1 and is_found:
        var = i_var

    return var, is_found

八

内置库

执行器中提供了丰富的内置库（Python 自带内置库、第三方库、平台自研内置库）来满足各种使用场景，可以在任何位置的脚本中导入使用。

九

总结

自动化执行器是自动化平台老方案中自动化框架的超集，提供框架能力的同时还负责解释执行自动化用例。技术实现上融入了优秀自动化框架的核心理念，也拓展支持了很多实用功能，同时更灵活、更具可塑性、更适配平台化，并支持“在线可视化、组件化(可复用)、全代码、低代码、零代码”编写用例，也能满足低门槛、高效率产出自动化用例的需要，还能实现极其复杂的测试场景。在自动化平台转型后的近两年时间里，平台在质量保障、效率提升和使用体验方面效果显著，后续我们也将持续打磨完善执行器和自动化平台。

---

往期回顾

---

*文/千仞

关注得物技术，每周一、三、五更新技术干货
要是觉得文章对你有帮助的话，欢迎评论转发点赞~
未经得物技术许可严禁转载，否则依法追究法律责任。

“

扫码添加小助手微信

如有任何疑问，或想要了解更多技术资讯，请添加小助手微信：